package com.enjoyor.soa.traffic.frame.support.amq;

import java.util.concurrent.TimeUnit;

import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

import com.enjoyor.soa.traffic.util.helper.LoggerHelper;

/**
 * ActiveMQ 消息发送器，负责向 AMQ 消息引擎发送消息，使用代码示例：
 * <p>
 *
 * <pre>
 * sender = new AMQSender(url, queue);
 * sender.initialize();
 * sender.sendMessage(&quot;message&quot;);
 * // <b>[注意]</b> 由于消息发送基于事务模式，务必在发送后进行提交
 * sender.commit();
 * sender.endEngine();
 * </pre>
 *
 * </p>
 *
 * @author WY
 * @reviser WY
 * @date 2015-4-8
 */

public class AMQSender extends AbstractActiveMQ {
	// MessageProducer：消息发送者
	private MessageProducer producer;
	// 消息持久化时间
	private long persistentTime;

	/**
	 * 生成一个非持久化消息发送器
	 *
	 * @param url
	 *            消息发送器URL
	 * @param queue
	 *            消息发送器名称
	 */
	public AMQSender(String url, String queue) {
		super(url, queue);
		setDeliveryMode(0);
	}

	/**
	 * 生成一个持久化消息发送器
	 *
	 * @param url
	 *            消息发送器URL
	 * @param queue
	 *            消息发送器名称
	 */
	public AMQSender(String url, String queue, long time) {
		super(url, queue);
		setDeliveryMode(time);
	}

	/**
	 * 根据消息持久化时间设置持久化模式，如果持久化时间为0， 则为非持久化模式。
	 *
	 * @param persistentTime
	 *            消息持久化时间
	 */
	private void setDeliveryMode(long persistentTime) {
		this.persistentTime = persistentTime;
	}

	@Override
	public void initialize() {
		if (this.running()) {
			try {
				// ConnectionFactory ：连接工厂，JMS 用它创建连接
				ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
						getUrl());
				// 构造从工厂得到连接对象
				setConnection(connectionFactory.createConnection());
				// 启动
				getConnection().start();
				// 获取操作连接
				setSession(getConnection().createSession(true,
						Session.AUTO_ACKNOWLEDGE));
				// 获取Destination
				Destination destination = getSession().createQueue(getQueue());
				// 得到消息生成者
				producer = getSession().createProducer(destination);
				// 设置持久化
				setMessageDeliveryMode();
			} catch (Exception e) {
				LoggerHelper.AMQ_LOG.warn(getQueue()
						+ " AMQ消息发送引擎连接异常，正在重新连接...");
				try {
					TimeUnit.SECONDS.sleep(CONNECTION_FAILED_DELAY);
				} catch (InterruptedException e1) {
					LoggerHelper.AMQ_LOG
							.warn(getQueue() + " AMQ消息发送引擎延时异常", e1);
				}
				initialize();
			}
		}
	}

	private void setMessageDeliveryMode() throws JMSException {
		if (this.persistentTime == 0) {
			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		} else {
			producer.setTimeToLive(60 * 60 * 1000);
		}
	}

	/**
	 * 发送一条消息
	 *
	 * @param message
	 *            需要发送的消息
	 */
	public void sendMessage(String message) {
		try {
			producer.send(getSession().createTextMessage(message));
		} catch (JMSException e) {
			LoggerHelper.LOG.error(getQueue() + " AMQ消息发送异常，正在准备关闭发送引擎......",
					e);
			rebootSender();
		}
	}

	/**
	 * 提交消息事务
	 */
	public void commit() {
		try {
			getSession().commit();
		} catch (JMSException e) {
			LoggerHelper.LOG.error(getQueue() + " AMQ事务提交异常，正在准备关闭发送引擎......",
					e);
			rebootSender();
		}
	}

	private void rebootSender() {
		endSender();
		LoggerHelper.LOG.info(getQueue() + " AMQ消息发送引擎已经处于关闭状态，正在准备重新启动......");
		initialize();
	}

	private void endSender() {
		try {
			producer.close();
			getSession().close();
			getConnection().close();
		} catch (JMSException e) {
			LoggerHelper.AMQ_LOG.error(getQueue() + " AMQ消息发送引擎关闭异常", e);
		}
	}

	@Override
	public void endEngine() {
		endSender();
		this.shutdown();
	}

	@Override
	public Message receiveMessage(int time) {
		// TODO Auto-generated method stub
		return null;
	}
}
